package com.swak.rabbit;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import com.swak.rabbit.RabbitMQTemplate.MessageHandler;
import com.swak.rabbit.connection.CacheChannelProxy;
import com.swak.rabbit.message.Message;
import com.swak.reactivex.threads.Context;
import com.swak.reactivex.threads.Contexts;
import com.swak.utils.StringUtils;

/**
 * 具有自动连接功能<br>
 * 
 * 业务返回失败，进入消费重试<br>
 * 业务执行异常，进入死信队列， 最好为每个队列配置死信队列<br>
 * 消息只能按照顺序一个一个的处理，如果这个消费者在处理一个消息，没有应答，则不会处理下一个消息。<br>
 * 虽然支持异步消费，但是此消息每处理完之前不会处理下一个消息<br>
 * 
 * 
 * 第212行代码：设置不自动应答 ，且每次只消费一条数据 <br>
 * 
 * @author lifeng
 */
public class TemplateConsumer implements Consumer {

	private final Logger logger = LoggerFactory.getLogger(TemplateConsumer.class);
	private final String queue;
	private final String retry;
	private final RabbitMQTemplate template;
	private CacheChannelProxy channel;
	private MessageHandler messageHandler;
	private final int prefetch;
	private String consumerTag;

	public static TemplateConsumer of(String queue, String retry, int prefetch, RabbitMQTemplate template,
			MessageHandler messageHandler) throws IOException, TimeoutException {
		return new TemplateConsumer(queue, retry, prefetch, template, messageHandler);
	}

	private TemplateConsumer(String queue, String retry, int prefetch, RabbitMQTemplate template,
			MessageHandler messageHandler) throws IOException, TimeoutException {
		this.prefetch = prefetch;
		this.queue = queue;
		this.retry = retry;
		this.template = template;
		this.messageHandler = messageHandler;
		this.resetChannel();
	}

	@Override
	@SuppressWarnings("unchecked")
	public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
			throws IOException {
		// 当前的执行环境
		Context context = Contexts.currentContext();
		Message message = null;
		try {
			message = this.buildMessage(envelope, properties, body);
			Object result = this.messageHandler.handle(message);
			if (result != null && result instanceof CompletionStage) {
				this.asynHandleResult(context, (CompletionStage<Object>) result, envelope, message);
			} else {
				this.handleResult(result, envelope, message, null);
			}
		} catch (Throwable e) {
			this.handleError(envelope, message, e);
			logger.error("Consume Message error:", e);
		}
	}

	// 创建消息
	private Message buildMessage(Envelope envelope, BasicProperties properties, byte[] body) {
		Message message = Message.of().setProperties(properties).setPayload(body);
		if (Constants.isRetryMessage(envelope.getExchange())) {
			message.retryMessage();
		}
		return this.debugMessage(message);
	}

	// 调试信息
	@SuppressWarnings("unchecked")
	private Message debugMessage(Message message) {
		if (logger.isDebugEnabled() || logger.isInfoEnabled()) {
			Object retryQueue = null;
			int retryCount = 0;
			if (message.getProperties().getHeaders() != null) {
				List<Map<String, Object>> deaths = (List<Map<String, Object>>) message.getProperties().getHeaders()
						.get("x-death");
				if (deaths != null && deaths.size() > 0) {
					retryQueue = deaths.get(0).get("queue");
					for (Map<String, Object> death : deaths) {
						if (String.valueOf(death.get("reason")).equals("expired")) {
							retryCount++;
						}
					}
				}
			}
			message.setRetryQueue(String.valueOf(retryQueue)).setConsumeRetrys(retryCount);
		}
		return message;
	}

	// 异步处理消息回调
	private void asynHandleResult(Context context, CompletionStage<Object> resultFuture, Envelope envelope,
			Message message) {
		if (context != null) {
			resultFuture.whenComplete((v, e) -> {
				context.runOnContext(() -> {
					this.handleResult(v, envelope, message, e);
				});
			});
		} else {
			resultFuture.whenComplete((v, e) -> {
				this.handleResult(v, envelope, message, e);
			});
		}
	}

	// 处理消息回调
	private void handleResult(Object result, Envelope envelope, Message message, Throwable ex) {
		try {
			if (ex != null || (result != null && result instanceof Boolean && !(Boolean) result)) {
				this.handleError(envelope, message, ex);
			} else {
				this.handleSuccess(envelope, message);
			}
		} catch (Exception e) {
			logger.error("Consumer Ack error:", e);
		}
	}

	// 消息成功
	private void handleSuccess(Envelope envelope, Message message) throws IOException {
		try {
			if (this.channel.isOpen()) {
				this.channel.basicAck(envelope.getDeliveryTag(), false);
			}
			if (logger.isDebugEnabled()) {
				if (StringUtils.isBlank(message.getOrigin())) {
					logger.debug("Consume Queue[{}] - Message[{}] Success.", queue, message.getId());
				} else {
					logger.debug("Consume Queue[{}] - Message[{}] - Origin[{}] - Retry[{}] - Times[{}] Success.", queue,
							message.getId(), message.getOrigin(), message.getRetryQueue(), message.getConsumeRetrys());
				}
			} else if (logger.isInfoEnabled()) {
				if (!StringUtils.isBlank(message.getOrigin())) {
					logger.info("Consume Queue[{}] - Message[{}] - Origin[{}] - Retry[{}] - Times[{}] Success.", queue,
							message.getId(), message.getOrigin(), message.getRetryQueue(), message.getConsumeRetrys());
				}
			}
		} catch (Exception e) {
			logger.error("Consumer Ack error:", e);
		}
	}

	// 消息失败
	private void handleError(Envelope envelope, Message message, Throwable ex) throws IOException {
		try {

			if (ex != null) {
				logger.error("Consume Queue[{}] Error {}.", queue, ex);
			}

			if (this.channel.isOpen()) {
				try {
					this.channel.basicNack(envelope.getDeliveryTag(), false, false);
				} catch (Exception e) {
					this.channel.basicNack(envelope.getDeliveryTag(), false, false);
				}
			}
			if (logger.isDebugEnabled() || logger.isInfoEnabled()) {
				if (StringUtils.isBlank(message.getOrigin())) {
					logger.debug("Consume Queue[{}] - Message[{}] Error {}.", queue, message.getId(), ex);
				} else {
					logger.debug("Consume Queue[{}] - Message[{}] - Origin[{}] - Retry[{}] - Times[{}] Error {}.",
							queue, message.getId(), message.getOrigin(), message.getRetryQueue(),
							message.getConsumeRetrys(), ex);
				}
			}
		} catch (Exception e) {
			logger.error("Consumer Ack error:", e);
		}
	}

	// 充分利用已经存在的通道
	private synchronized void resetChannel() throws IOException, TimeoutException {
		if (this.channel != null && this.channel.isOpen()) {
			this.channel.physicalClose();
		}
		this.channel = template.forChannel();

		/*
		 * 这是主动注册回调，对于业务来说是推模式
		 */
		this.channel.basicQos(prefetch);

		/*
		 * 第二个参数设置是否自动确认
		 */
		this.consumerTag = this.channel.basicConsume(queue, false, this);

		/*
		 * 这是主动的获取数据，对于业务来说是拉模式
		 */
		// this.channel.basicGet(arg0, arg1);

		/*
		 * 使用默认的自动重试
		 */
		RabbitMQTemplate.putMessageHandler(this.queue, this.messageHandler);
		if (StringUtils.isNotBlank(this.retry)) {
			RabbitMQTemplate.putMessageHandler(this.retry, this.messageHandler);
		}
	}

	@Override
	public void handleShutdownSignal(String consumerTag, ShutdownSignalException cause) {
		logger.warn("Consumer for queue {} Shutdown", this.queue);
	}

	///////////// 如下的事件不用处理 /////////////////////////////////////
	@Override
	public void handleConsumeOk(String consumerTag) {
	}

	@Override
	public void handleCancelOk(String consumerTag) {
	}

	@Override
	public void handleCancel(String consumerTag) throws IOException {
	}

	@Override
	public void handleRecoverOk(String consumerTag) {
		logger.error("Consumer Recover OK:", consumerTag);
	}

	public void cancel() {
		try {
			if (StringUtils.isBlank(this.consumerTag)) {
				this.channel.basicCancel(consumerTag);
			}
		} catch (IOException e) {
		}
	}
}